package com.zxl.support;

import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;

/**
 * @Description: 数据量统计线程池管理工具类
 *
 * @Author:zhaoxl1
 *
 * @Date: 2022/9/23
 *
 */
@Component
@Slf4j
@Data
public class ExecutorThreadPool {

    private ConcurrentHashMap<String, ThreadPoolExecutor> executorMap;

    private Semaphore semaphore = new Semaphore(1);

    public ExecutorThreadPool() {
        this.executorMap = new ConcurrentHashMap<>();
    }

    /**
     * @Param: null
     *
     * @return: null
     *
     * @Description:根据表所属数据库id创建线程池，并装配入ConcurrentHashMap
     *
     */
    private void setExecutorMap() {
        List<Integer> metaDatabaseIds = new ArrayList<>();
        metaDatabaseIds.add(11);
        metaDatabaseIds.add(12);
        for (Integer metaDatabaseId : metaDatabaseIds) {
            addExecutor("database_" + metaDatabaseId);
        }
    }

    private void addExecutor(String key) {
        if (!executorMap.containsKey(key)) {
            ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
                    5,
                    5,
                    3L,
                    TimeUnit.SECONDS,
                    new LinkedBlockingQueue<>(10000),
                    Executors.defaultThreadFactory());
            executorMap.put(key, threadPoolExecutor);
        }
    }

    /**
     * @Param: datasourceId
     *         数据库id
     *
     * @return: java.util.concurrent.ThreadPoolExecutor
     *          线程池
     *
     * @Description:根据数据库id获取线程池
     *
     */
    public ThreadPoolExecutor getThreadPoolExecutor(String datasourceId) {
        setExecutorMap();
        if (!executorMap.containsKey(datasourceId)) {
            while (!executorMap.containsKey(datasourceId) && !semaphore.tryAcquire()) {
                try {
                    TimeUnit.MILLISECONDS.sleep(50);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
            try {
                addExecutor(datasourceId);
            } finally {
                semaphore.release();
            }
        }
        return executorMap.get(datasourceId);
    }
}
